package org.citywithincity.jobqueue;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.citywithincity.models.Models;
import com.damai.plugins.IPlugin;


public class TaskQueue implements IPlugin,JobQueueManager, QueueListener {
	private static TaskQueue instance = new TaskQueue();
	private static PcQueue jobQueue;
	
	private static final Log logger  = LogFactory.getLog(TaskQueue.class);

	
	public static final String DEFAULT = "_defaultAsyncJob";
	
	/**
	 * 
	 * 目前有4中执行异步任务的方式
	 * 1、setHandler预先设置entity对应的处理器，然后用add执行
	 * 2、使用registerAsyncJobFactory注册一个AsyncJobFactory,然后使用execute执行
	 * 本方式可以使用name区分不同的业务逻辑
	 * 3、直接执行execute(),需要一个监听器和一个传入参数
	 * 4、使用Aop,AsyncInterceptor注解方法，并使用model调用
	 * 
	 * 
	 * 
	 * @author renxueliang
	 *
	 */
	public static TaskQueue getInstance() {
		return instance;
	}

	private TaskQueue() {
		registerAsyncJobFactory(DEFAULT, new DefaultAsyncJobFactory());
	}
	
	private Map<String, AsyncJobFactory> map = new ConcurrentHashMap<String, AsyncJobFactory>();
	
	/**
	 * 注册异步任务的数据工厂
	 * @param name
	 * @param factory
	 */
	public void registerAsyncJobFactory(String name,AsyncJobFactory factory){
		try {
			Models.assignModels(factory);
		} catch (Exception e) {
			throw new RuntimeException("assignModels失败"+factory.getClass(),e);
		}
		map.put(name, factory);
	}
	
	public void execute(String name,Object data){
		AsyncJobFactory factory = map.get(name);
		if(factory==null){
			throw new RuntimeException("没有注册工厂"+name);
		}
		add(factory.createJob(data));
	}
	
	public static class DefaultAsyncJobFactory implements AsyncJobFactory{

		@SuppressWarnings("unchecked")
		@Override
		public AsyncJob createJob(Object data) {
			Map<String, Object> map = (Map<String, Object>)data;
			AsyncJobListener<?> listener = (AsyncJobListener<?>) map.get("listener");
			Object param = map.get("param");
			return new DefaultAsyncJob(listener,param);
		}
		
	}
	
	public static class DefaultAsyncJob implements AsyncJob{

		@SuppressWarnings("rawtypes")
		private AsyncJobListener listener;
		private Object data;
		public DefaultAsyncJob(AsyncJobListener<?> listener,Object data){
			this.listener = listener;
			this.data = data;
		}
		
		@SuppressWarnings("unchecked")
		@Override
		public void execute() {
			listener.execute(data);
		}
		
	}

	public <T> void execute(AsyncJobListener<T> listener,T data){
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("listener", listener);
		map.put("param", data);
		execute(DEFAULT, map);
	}
	
	public void setThreadCount(int count){
		jobQueue = new PcQueue();
		jobQueue.setThreadCount(count);
		jobQueue.setListener(this);
	}

	@Override
	public void start() {
		jobQueue.start();
	}

	@Override
	public void stop() {
		jobQueue.stop();
	}

	@Override
	public void add(Object job) {
		jobQueue.add(job);
		logger.info("数据插入异步队列成功" +job );
	}

	@Override
	public void onAddJob(Object job) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void onRemoveJob(Object job) {
		// TODO Auto-generated method stub
		
	}
	
	
	public JobHandler<?> getHandler(Class<?> clazz){
		return jobQueue.getHandler(clazz);
	}

	@Override
	public <T> void setHandler(Class<T> clazz, JobHandler<T> handler) {
		try {
			Models.assignModels(handler);
		} catch (Exception e) {
			throw new RuntimeException("不能设置模型",e);
		}
		jobQueue.setHandler(clazz, handler);
	}


}
